package io.kiki.sba.registry.server.clientInterface.converter;


import io.kiki.sba.registry.common.model.DataCenterPushInfo;
import io.kiki.sba.registry.common.model.SegmentPushInfo;
import io.kiki.sba.registry.common.model.ServerData;
import io.kiki.sba.registry.common.model.metaserver.ProvideData;
import io.kiki.sba.registry.common.model.store.*;
import io.kiki.sba.registry.core.model.*;
import io.kiki.sba.registry.exception.SofaRegistryRuntimeException;
import io.kiki.sba.registry.util.DatumVersionUtil;
import io.kiki.sba.registry.util.StringFormatter;
import io.kiki.sba.registry.util.StringUtil;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.Map.Entry;
import java.util.function.Predicate;

/**
 * The type Received data converter.
 */
public final class ReceivedDataConverter {
    private static final Logger logger = LoggerFactory.getLogger(ReceivedDataConverter.class);

    private ReceivedDataConverter() {
    }


    public static PushData<ReceivedData> getReceivedData(DataCenterToSubDatumMap dataCenterToSubDatumMap, Scope scope, List<String> subscriberRegisterIdList, String regionLocal, String localDataCenter, Predicate<String> pushdataPredicate) {

        if (null == dataCenterToSubDatumMap || CollectionUtils.isEmpty(dataCenterToSubDatumMap.getDataCenterToSubDatumMap())) {
            return new PushData<>(null, Collections.EMPTY_MAP);
        }
        dataCenterToSubDatumMap.mustUnzipped();
        ReceivedData receivedData = new ReceivedData();
        receivedData.setDataId(dataCenterToSubDatumMap.getDataId());
        receivedData.setGroup(dataCenterToSubDatumMap.getGroup());
        receivedData.setInstanceId(dataCenterToSubDatumMap.getInstanceId());
        receivedData.setSubscriberRegistIds(subscriberRegisterIdList);
        receivedData.setScope(scope.name());
        receivedData.setLocalZone(regionLocal);

        DataCenterPushInfo dataCenterPushInfo = fillRegionDatas(dataCenterToSubDatumMap, receivedData, localDataCenter, pushdataPredicate);

        return new PushData<>(receivedData, Collections.singletonMap(receivedData.getSegment(), dataCenterPushInfo));
    }

    public static PushData<MultiReceivedData> getMultiReceivedData(DataCenterToSubDatumMap unzipDatum, Scope scope, List<String> subscriberRegisterIdList, String regionLocal, String localDataCenter, Predicate<String> pushdataPredicate, Map<String, Set<String>> segmentZones) {
        if (null == unzipDatum || CollectionUtils.isEmpty(unzipDatum.getDataCenterToSubDatumMap())) {
            return new PushData<>(null, Collections.EMPTY_MAP);
        }
        unzipDatum.mustUnzipped();

        MultiReceivedData multiReceivedData = new MultiReceivedData();
        multiReceivedData.setDataId(unzipDatum.getDataId());
        multiReceivedData.setGroup(unzipDatum.getGroup());
        multiReceivedData.setInstanceId(unzipDatum.getInstanceId());
        multiReceivedData.setSubscriberRegistIds(subscriberRegisterIdList);
        multiReceivedData.setScope(scope.name());
        multiReceivedData.setLocalZone(regionLocal);
        multiReceivedData.setLocalSegment(localDataCenter);

        Map<String, DataCenterPushInfo> dataCenterPushInfo = fillMultiRegionData(unzipDatum, localDataCenter, multiReceivedData, pushdataPredicate, segmentZones);
        return new PushData<>(multiReceivedData, dataCenterPushInfo);
    }

    private static Map<String, DataCenterPushInfo> fillMultiRegionData(DataCenterToSubDatumMap dataCenterToSubDatumMap, String localDataCenter, MultiReceivedData multiReceivedData, Predicate<String> pushdataPredicate, Map<String, Set<String>> segmentZones) {

        final Map<String, MultiSegmentData> multiDatas = new HashMap<>();
        final Map<String, DataCenterPushInfo> dataCenterPushInfo = new HashMap<>();

        for (Entry<String, SubDatum> entry1 : dataCenterToSubDatumMap.getDataCenterToSubDatumMap().entrySet()) {
            String pushDataCenter = entry1.getKey();
            SubDatum subDatum = entry1.getValue();

            Map<String, SegmentDataCounter> multiSegmentDatas = buildMultiSegmentDataFromSubDatum(localDataCenter, pushDataCenter, subDatum, pushdataPredicate, segmentZones.get(pushDataCenter));
            if (CollectionUtils.isEmpty(multiSegmentDatas)) {
                continue;
            }

            final Map<String, SegmentPushInfo> segmentPushInfo = new HashMap<>();

            for (Entry<String, SegmentDataCounter> entry : multiSegmentDatas.entrySet()) {
                SegmentDataCounter segmentDataCounter = entry.getValue();
                multiDatas.put(entry.getKey(), segmentDataCounter.getSegmentData());
                segmentPushInfo.put(entry.getKey(), new SegmentPushInfo(entry.getKey(), segmentDataCounter.getDataCount()));
            }
            dataCenterPushInfo.put(pushDataCenter, new DataCenterPushInfo(subDatum.getVersion(), segmentPushInfo));
        }
        multiReceivedData.setMultiData(multiDatas);
        return dataCenterPushInfo;
    }

    private static Map<String, SegmentDataCounter> buildMultiSegmentDataFromSubDatum(String localDataCenter, String pushDataCenter, SubDatum subDatum, Predicate<String> pushdataPredicate, Set<String> segmentZones) {

        Map<String, SegmentDataCounter> map = new HashMap<>();
        if (StringUtils.equals(localDataCenter, pushDataCenter)) {
            LocalDataCenterPushData localDataCenterPushData = localSegmentData(localDataCenter, subDatum, pushdataPredicate, segmentZones);
            if (localDataCenterPushData == null) {
                return null;
            }

            map.put(localDataCenter, localDataCenterPushData.getLocalSegmentDatas());
            map.putAll(localDataCenterPushData.getRemoteSegmentDatas());
            return map;
        }
        return remoteSegmentData(subDatum, segmentZones);
    }

    private static Map<String, SegmentDataCounter> remoteSegmentData(SubDatum subDatum, Set<String> segmentZones) {

        Map<String, List<Data>> pushData;
        try {
            if (CollectionUtils.isEmpty(segmentZones)) {
                throw new SofaRegistryRuntimeException(StringFormatter.format("segmentZones is empty."));
            }
            pushData = swizzData(subDatum, null);
        } catch (Throwable th) {
            logger.error("build remoteSegmentData error, dataId: {}.", subDatum.getDataId(), th);
            return null;
        }

        Map<String, SegmentDataCounter> map = new HashMap<>(segmentZones.size());
        for (String zone : segmentZones) {
            SegmentDataCounter segmentDataCounter = new SegmentDataCounter(new MultiSegmentData(zone, subDatum.getVersion()));
            segmentDataCounter.put(zone, pushData.get(zone));
            map.put(zone, segmentDataCounter);
        }
        return map;
    }

    private static LocalDataCenterPushData localSegmentData(String localDataCenter, SubDatum subDatum, Predicate<String> pushdataPredicate, Set<String> segmentZones) {

        // split into local and remote
        Map<String, List<Data>> pushData;
        try {
            if (CollectionUtils.isEmpty(segmentZones)) {
                throw new SofaRegistryRuntimeException(StringFormatter.format("segmentZones is empty."));
            }
            pushData = swizzData(subDatum, null);
        } catch (Throwable th) {
            logger.error("build localSegmentData error, dataId: {}.", subDatum.getDataId(), th);
            return null;
        }

        LocalDataCenterPushData localDataCenterPushData = new LocalDataCenterPushData();
        localDataCenterPushData.from(pushData, localDataCenter, subDatum.getVersion(), pushdataPredicate, segmentZones);
        return localDataCenterPushData;
    }

    private static DataCenterPushInfo fillRegionDatas(DataCenterToSubDatumMap unzipDatum, ReceivedData receivedData, String localDataCenter, Predicate<String> predicate) {

        StringUtil.checkEquals(unzipDatum.dataCenterSet(), Collections.singleton(localDataCenter), "fillRegionDatas.dataCenter");
        SubDatum subDatum = unzipDatum.getSubDatum(localDataCenter);
        receivedData.setSegment(localDataCenter);
        receivedData.setVersion(subDatum.getVersion());

        Map<String, List<Data>> map = swizzData(subDatum, predicate);
        receivedData.setData(map);

        Map<String, Integer> pushDataCount = new HashMap<>(map.size());

        int dataCount = 0;
        for (Entry<String, List<Data>> entry : map.entrySet()) {
            int size = entry.getValue().size();
            pushDataCount.put(entry.getKey(), size);
            dataCount += size;
        }

        receivedData.setDataCount(pushDataCount);
        DataCenterPushInfo dataCenterPushInfo = new DataCenterPushInfo(localDataCenter, subDatum.getVersion(), new SegmentPushInfo(localDataCenter, dataCount));
        return dataCenterPushInfo;
    }

    private static Map<String /*zone*/, List<Data>> swizzData(SubDatum subDatum, Predicate<String> pushdataPredicate) {
        Map<String /*zone*/, List<Data>> swizzMap = new HashMap<>();
        List<SubPublisher> subPublisherList = subDatum.getUnzippedSubPublisherList();
        if (subPublisherList.isEmpty()) {
            return Collections.EMPTY_MAP;
        }
        for (SubPublisher subPublisher : subPublisherList) {
            List<ServerData> serverDataList = subPublisher.getServerDataList();

            String region = subPublisher.getCell();

            if (pushdataPredicate != null && pushdataPredicate.test(region)) {
                continue;
            }
            if (null == serverDataList) {
                serverDataList = new ArrayList<>();
            }
            List<Data> regionDatas = swizzMap.computeIfAbsent(region, k -> new ArrayList<>());
            fillRegionDatas(regionDatas, serverDataList);
        }
        return swizzMap;
    }

    private static void fillRegionDatas(List<Data> dataList, List<ServerData> serverDataList) {
        for (ServerData serverData : serverDataList) {
            Data data = new Data();
            try {
                String dataString = (String) serverData.extract();
                data.setData(dataString);
                dataList.add(data);
            } catch (Exception e) {
                logger.error("ReceivedData convert error", e);
            }
        }
    }

    public static ReceivedConfigData getReceivedConfigData(ServerData serverData, DataInfo dataInfo, Long version) {
        ReceivedConfigData receivedConfigData = new ReceivedConfigData();

        if (serverData != null) {
            Data data = new Data();
            String dataString = (String) serverData.getObject();
            data.setData(dataString);
            receivedConfigData.setData(data);
        }
        receivedConfigData.setDataId(dataInfo.getDataId());
        receivedConfigData.setGroup(dataInfo.getGroup());
        receivedConfigData.setInstanceId(dataInfo.getInstanceId());
        if (version == null) {
            version = DatumVersionUtil.nextId();
        }
        receivedConfigData.setVersion(DatumVersionUtil.transferDatumVersion(version));
        return receivedConfigData;
    }

    public static ReceivedConfigData createReceivedConfigData(Watcher watcher, ProvideData provideData) {
        DataInfo dataInfo = DataInfo.valueOf(watcher.getDataInfoId());
        ReceivedConfigData receivedConfigData = ReceivedDataConverter.getReceivedConfigData(provideData.getServerData(), dataInfo, provideData.getVersion());
        receivedConfigData.setConfiguratorRegistIds(new ArrayList<>(Arrays.asList(watcher.getRegisterId())));
        return receivedConfigData;
    }
}
